package com.run.fjy.io;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

public class TerminalWritable
	implements Writable
{

	private boolean isLocal;
	private long firstTime = Long.MAX_VALUE;
	private Map<Text,HashSet<Text>> global = new HashMap<Text,HashSet<Text>>();
	private Map<Text,HashSet<Text>> local  = new HashMap<Text,HashSet<Text>>();
	private Map<Text, Text> otherFields = new HashMap<Text,Text>();


	public String toString()
	{
		return (new StringBuilder()).append("TerminalWritable [isLocal=").append(isLocal).append(", firstTime=").append(firstTime).append(", global=").append(global).append(", local=").append(local).append(", otherFields=").append(otherFields).append("]").toString();
	}

	public Map<Text, Text> getOtherFields()
	{
		return otherFields;
	}

	public void setOtherFields(Map<Text, Text> otherFields)
	{
		this.otherFields = otherFields;
	}

	public long getFirstTime()
	{
		return firstTime;
	}

	public void setFirstTime(long firstTime)
	{
		this.firstTime = firstTime;
	}

	public boolean isLocal()
	{
		return isLocal;
	}

	public void setLocal(boolean isLocal)
	{
		this.isLocal = isLocal;
	}

	public void addLocal(Text name, Text value)
	{
		HashSet<Text> set = (HashSet<Text>)local.get(name);
		if (null == set)
			set = new HashSet<Text>();
		set.add(value);
		local.put(name, set);
	}

	public void addAllLocal(Text name, HashSet<Text> values)
	{
		HashSet<Text> set = (HashSet<Text>)local.get(name);
		if (null == set)
			set = new HashSet<Text>();
		set.addAll(values);
		local.put(name, set);
	}

	public Map<Text,HashSet<Text>> getAllLocal()
	{
		return local;
	}

	public void addGlobal(Text name, Text value)
	{
		HashSet<Text> set = (HashSet<Text>)global.get(name);
		if (null == set)
			set = new HashSet<Text>();
		set.add(value);
		global.put(name, set);
	}

	public void addAllGlobal(Text name, HashSet<Text> values)
	{
		HashSet<Text> set = (HashSet<Text>)global.get(name);
		if (null == set)
			set = new HashSet<Text>();
		set.addAll(values);
		global.put(name, set);
	}

	public Map<Text,HashSet<Text>> getAllGlobal()
	{
		return global;
	}

	public void write(DataOutput out)
		throws IOException
	{
		out.writeBoolean(isLocal);
		out.writeLong(firstTime);
		out.writeInt(global.size());
		for (Map.Entry<Text,HashSet<Text>> entry : global.entrySet())
		{
			entry.getKey().write(out);
			out.writeInt(entry.getValue().size());
			for(Text value : entry.getValue()) {
				value.write(out);
			}
		}

		out.writeInt(local.size());
		for (Map.Entry<Text,HashSet<Text>> entry : local.entrySet())
		{
			entry.getKey().write(out);
			out.writeInt(entry.getValue().size());
			for(Text value : entry.getValue()) {
				value.write(out);
			}
		}

		out.writeInt(otherFields.size());
		for (Map.Entry<Text,Text> entry : otherFields.entrySet())
		{
			entry.getKey().write(out);
			entry.getValue().write(out);
		}

	}

	public void readFields(DataInput in)
		throws IOException
	{
		isLocal = in.readBoolean();
		firstTime = in.readLong();
		for (int size = in.readInt(); size-- > 0;)
		{
			Text key = new Text();
			key.readFields(in);
			int setSize = in.readInt();
			HashSet<Text> set = new HashSet<Text>();
			while (setSize-- > 0) 
			{
				Text setValue = new Text();
				setValue.readFields(in);
				set.add(setValue);
			}
			global.put(key, set);
		}

		for (int size = in.readInt(); size-- > 0;)
		{
			Text key = new Text();
			key.readFields(in);
			int setSize = in.readInt();
			HashSet<Text> set = new HashSet<Text>();
			while (setSize-- > 0) 
			{
				Text setValue = new Text();
				setValue.readFields(in);
				set.add(setValue);
			}
			local.put(key, set);
		}

		for (int size = in.readInt(); size-- > 0;)
		{
			Text key = new Text();
			Text value = new Text();
			key.readFields(in);
			value.readFields(in);
			otherFields.put(key, value);
		}

	}

	public void clear()
	{
		global.clear();
		local.clear();
		otherFields.clear();
	}
}